import json
import os
from datetime import datetime,timedelta
import pandas as pd
from app import pro
from app.config import Config
import numpy as np
import duckdb
from app.core.stock_utils import stockUtils
import qstock as qs
from typing import Optional

class StockMonitorService:

    def __init__(self):
        self.watchlist = {}
        self.cache_file = os.path.join(Config.BASE_DIR, "stock_cache.json")
        self.load_watchlist()
    
    def load_watchlist(self):
        try:
            if os.path.exists(Config.CONFIG_FILE):
                with open(Config.CONFIG_FILE, 'r', encoding='utf-8') as f:
                    data = json.load(f)
                    self.watchlist = data.get('watchlist', {})
        except Exception as e:
            print(f"Error loading watchlist: {str(e)}")
            self.watchlist = {}
    
    #获取监控股票基本信息及交易数据
    def get_stock_info(self, stock_code: str):

        con = duckdb.connect(database='db/stock_prices.duckdb', read_only=False)
        #从stock_info表获取所有数据
        sql = f'''
        SELECT * FROM stock_info WHERE ts_code='{stock_code}'
        '''
        df = con.execute(sql).df()
        stock_info = {
            'ts_code': stock_code,
            'stock_name': df['stock_name'].values[0] if not df['stock_name'].isna().all() else None,
            'industry': df['industry'].values[0] if not df['industry'].isna().all() else None,
            'gap': df['gap'].values[0] if not df['gap'].isna().all() else None,
            'out_funds': float(round(df['out_funds'].values[0],2)) if not df['out_funds'].isna().all() else None,
            'domestic': float(round(df['domestic'].values[0],2)) if not df['domestic'].isna().all() else None,
            'undervalued': float(round(df['undervalued'].values[0],2)) if not df['undervalued'].isna().all() else None,
            'reasonable': float(round(df['reasonable'].values[0],2)) if not df['reasonable'].isna().all() else None,
            'overvalued': float(round(df['overvalued'].values[0],2)) if not df['overvalued'].isna().all() else None,
            'crazy': float(round(df['crazy'].values[0],2)) if not df['crazy'].isna().all() else None,
        }
        #stock_info['out_funds']进行四舍五入保留2位小数
        stock_info['out_funds'] = float(round(stock_info['out_funds'],2)) if stock_info['out_funds'] is not None else None
        stock_info['domestic'] = float(round(stock_info['domestic'],2)) if stock_info['domestic'] is not None else None
        stock_info['undervalued'] = float(round(stock_info['undervalued'],2)) if stock_info['undervalued'] is not None else None
        stock_info['reasonable'] = float(round(stock_info['reasonable'],2)) if stock_info['reasonable'] is not None else None
        stock_info['overvalued'] = float(round(stock_info['overvalued'],2)) if stock_info['overvalued'] is not None else None
        stock_info['crazy'] = float(round(stock_info['crazy'],2)) if stock_info['crazy'] is not None else None

        con.close()
        # 关闭数据库连接
        return stock_info
    
    def get_watchlist(self):
        result = []
        con = duckdb.connect(database='db/stock_prices.duckdb', read_only=False)
        #从stock_info表获取所有数据
        sql = '''
        SELECT a.*,c.close,c.pct_change FROM stock_info a,
        (select ts_code,MAX(date) date from stock_data GROUP by ts_code) b,
        stock_data c
        where a.ts_code=b.ts_code and b."date" =c.date and b.ts_code=c.ts_code
        '''
        df = con.execute(sql).df()
        con.close()
        
        # 处理 NaN 和 Infinity 值
        # 首先替换无穷大值
        df = df.replace([np.inf, -np.inf], np.nan)
        # 然后填充 NaN 值
        # df = df.fillna(value=None)
        df = df.where(pd.notnull(df), None)
        
        # 确保所有数据类型都是 JSON 兼容的
        for _, row in df.iterrows():
            item = {
                'stock_info': {
                    'ts_code': str(row['ts_code']) if pd.notna(row['ts_code']) else None,
                    'stock_name': str(row['stock_name']) if pd.notna(row['stock_name']) else None,
                    'industry': str(row['industry']) if pd.notna(row['industry']) else None,
                    'gap': str(row['gap']) if pd.notna(row['gap']) else None,
                    'out_funds': float(round(row['out_funds'],2)) if pd.notna(row['out_funds']) and row['out_funds'] is not None else None,
                    'domestic': float(round(row['domestic'],2)) if pd.notna(row['domestic']) and row['domestic'] is not None else None,
                    'undervalued': float(round(row['undervalued'],2)) if pd.notna(row['undervalued']) and row['undervalued'] is not None else None,
                    'reasonable': float(round(row['reasonable'],2)) if pd.notna(row['reasonable']) and row['reasonable'] is not None else None,
                    'overvalued': float(round(row['overvalued'],2)) if pd.notna(row['overvalued']) and row['overvalued'] is not None else None,
                    'crazy': float(round(row['crazy'],2)) if pd.notna(row['crazy']) and row['crazy'] is not None else None,
                    'bg_rate': str(row['bg_rate']) if pd.notna(row['bg_rate']) else None,
                    'close': float(round(row['close'],2)) if pd.notna(row['close']) and row['close'] is not None else None,
                    'pct_change': str(row['pct_change']) if pd.notna(row['pct_change']) else None
                }
            }
            result.append(item)
        # 关闭数据库连接
        return result
    
    async def get_gap_watchlist(self):
        result = []
        con = duckdb.connect(database='db/stock_prices.duckdb', read_only=False)
        #从gap_stock表获取status=1的所有数据
        sql = 'SELECT ts_code,stock_name,ann_date,gap_date,gap_change,p_change,summary,change_reason,type FROM gap_stock WHERE status=1'
        df = con.execute(sql).df()
        con.close()
        # 关闭数据库连接
        for _, row in df.iterrows():
            item = {
                'gap_stock': {
                    'ts_code': str(row['ts_code']) if pd.notna(row['ts_code']) else None,
                    'stock_name': str(row['stock_name']) if pd.notna(row['stock_name']) else None,
                    'ann_date': str(row['ann_date']) if pd.notna(row['ann_date']) else None,
                    'gap_date': str(row['gap_date']) if pd.notna(row['gap_date']) else None,
                    'gap_change': float(row['gap_change']) if pd.notna(row['gap_change']) and row['gap_change'] is not None and str(row['gap_change']).strip() != '' else None,
                    'p_change': str(row['p_change']) if pd.notna(row['p_change']) else None,
                    'summary': str(row['summary']) if pd.notna(row['summary']) else None,
                    'change_reason': str(row['change_reason']) if pd.notna(row['change_reason']) else None,
                }
            }
            result.append(item)

        return result
    
    async def delete_stock(self,stock_code: str):
        con = duckdb.connect(database='db/stock_prices.duckdb', read_only=False)
        #删除stock_info表中ts_code为stock_code的数据
        delete_sql = 'DELETE FROM stock_info WHERE ts_code = ?'
        con.execute(delete_sql, [stock_code])
        #删除stock_data表中ts_code为stock_code的数据
        delete_sql = 'DELETE FROM stock_data WHERE ts_code = ?'
        con.execute(delete_sql, [stock_code])
        con.close()
        return {"status": "success"}
    
    def add_watch(self, stock_code: str, industry: str = None, gap: str = None, foreign: float = None, domestic: float = None, undervalued: float = None, reasonable: float = None, overvalued: float = None, crazy: float = None):
        con = duckdb.connect(database='db/stock_prices.duckdb', read_only=False)
        
        stock_name = stockUtils.get_stock_name(stock_code)
        #获取营收和净利润增速
        bg_rate = stockUtils.get_bg_rate(stock_code)
        #将值插入到duckdb中
        insert_sql = 'INSERT INTO stock_info ("ts_code", "stock_name", "industry", "gap", "out_funds", "domestic", "undervalued", "reasonable", "overvalued", "crazy", "bg_rate") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)'
        con.execute(insert_sql, [stock_code, stock_name, industry, gap, foreign, domestic, undervalued, reasonable, overvalued, crazy, bg_rate])
        con.close()
        # 关闭数据库连接
        return {"status": "success"}
    
    def add_excel(self, df: pd.DataFrame):
        # 直接将df插入到duckdb的stock_info表中
        con = duckdb.connect(database='db/stock_prices.duckdb', read_only=False)
        try:
            # 获取df中第一列所有的值（股票名称）
            stock_names = df.iloc[:, 0].tolist()
            # 获取股票代码
            stock_codes_names = stockUtils.get_stock_code(stock_names)
            
            if stock_codes_names:
                # stock_codes_names是一个二维列表，每个元素是[股票代码，股票名称]
                # df中第一列的值是股票名称
                # 需要stock_codes_names中的股票名称与df的第一列的值进行匹配，将匹配到的股票代码添加到df的第一列
                
                # 创建股票名称到股票代码的映射
                name_to_code = {name: code for code, name in stock_codes_names}
                
                # 获取df第一列的股票名称
                df_stock_names = df.iloc[:, 0]
                
                # 匹配股票代码
                matched_codes = []
                for name in df_stock_names:
                    if name in name_to_code:
                        matched_codes.append(name_to_code[name])
                    else:
                        matched_codes.append(None)  # 如果没有匹配到，添加None
                
                # 将股票代码插入到df的第一列
                df.insert(0, 'ts_code', matched_codes)
                #遍历df中ts_code列，获取每个股票的bg_rate
                for index, row in df.iterrows():
                    stock_code = row['ts_code']
                    if stock_code:
                        bg_rate = stockUtils.get_bg_rate(stock_code)
                        df.at[index, 'bg_rate'] = bg_rate
                
                # 将'bg_rate'列移动到最后一列
                cols = list(df.columns)
                cols.remove('bg_rate')
                cols.append('bg_rate')
                df = df[cols]
            
            # 注册临时表并插入数据
            con.register('temp_df', df)
            #先删除stock_info表中与temp_df表中ts_code重复的数据
            con.execute('DELETE FROM stock_info WHERE ts_code IN (SELECT ts_code FROM temp_df)')
            # 执行插入操作
            con.execute('INSERT INTO stock_info SELECT * FROM temp_df')

            # 提交事务
            con.commit()
        except Exception as e:
            # 如果出现错误，回滚事务
            con.execute("ROLLBACK")
            raise e
        finally:
            con.close()
        # 关闭数据库连接
        return {"status": "success"}
    
    async def get_data(self):
        con = duckdb.connect(database='db/stock_prices.duckdb', read_only=False)
        
        # 1. 从stock_info表获取所有股票代码ts_code
        stock_info_df = con.execute('SELECT ts_code FROM stock_info').df()
        
        if not stock_info_df.empty:
            for _, row in stock_info_df.iterrows():
                ts_code = row['ts_code']
                
                # 2. 遍历ts_code，每个代码都去stock_data表中查找数据
                # 检查stock_data表中是否有该股票的数据
                query = "SELECT date FROM stock_data WHERE ts_code = ? ORDER BY date DESC LIMIT 1"
                result = con.execute(query, [ts_code]).fetchone()
                
                # 3. 判断数据情况并处理
                if result is None:
                    # 如果没有数据，则使用qs.get_data(ts_code,start='20240601')获取历史数据
                    try:
                        df = stockUtils.get_stock_data(ts_code)
                        if not df.empty:
                            # 将数据插入到stock_data表中
                            con.register('temp_df', df)
                            con.execute('INSERT INTO stock_data SELECT * FROM temp_df')
                            con.unregister('temp_df')
                    except Exception as e:
                        print(f"Error getting historical data for {ts_code}: {str(e)}")
                        continue
                else:
                    # 如果有数据，则判断是否是当天，如果不是当天则从该股票在表中最后一天的数据到今天为止的数据用get_data取出来
                    last_date = result[0]  # 最后一天的数据日期
                    today = datetime.now().strftime('%Y%m%d')
                    
                    if last_date != today:
                        try:
                            # 取该股票近100天的数据
                            exit_df = con.execute(f"SELECT date,ts_code,open,high,low,close,vol,turnover,turnover_rate FROM stock_data WHERE ts_code = '{ts_code}' ORDER BY date desc LIMIT 100").df()
                            
                            #将last_date日期加一天，并转换成字符串
                            last_date = (datetime.strptime(last_date, '%Y%m%d') + timedelta(days=1)).strftime('%Y%m%d')
                            # 从最后一天到今天的数据
                            df = stockUtils.get_stock_data(ts_code, start_date=last_date,exit_df=exit_df)
                                
                            if not df.empty:
                                # 将数据插入到stock_data表中
                                con.register('temp_df', df)
                                con.execute(f'INSERT INTO stock_data SELECT * FROM temp_df where date >= \'{last_date}\'')
                                con.unregister('temp_df')
                        except Exception as e:
                            print(f"Error getting incremental data for {ts_code}: {str(e)}")
                            continue
                # 如果有当日数据，则不需要取
        
        con.close()
        return {"status": "success"}

    async def update_stock_info(self, stock_code: str, industry: Optional[str] = None, gap: Optional[str] = None, out_funds: Optional[float] = None, domestic: Optional[float] = None, undervalued: Optional[float] = None, reasonable: Optional[float] = None, overvalued: Optional[float] = None, crazy: Optional[float] = None):
        con = duckdb.connect(database='db/stock_prices.duckdb', read_only=False)
        query = """
        UPDATE stock_info
        SET industry = ?, gap = ?, out_funds = ?, domestic = ?, undervalued = ?, reasonable = ?, overvalued = ?, crazy = ?
        WHERE ts_code = ?
        """
        con.execute(query, [industry, gap, out_funds, domestic, undervalued, reasonable, overvalued, crazy, stock_code])
        con.commit()
        con.close()
        return {"status": "success"}
    
    async def stock_select(self, strategy: str):
        con = duckdb.connect(database='db/stock_prices.duckdb', read_only=False)
        query = f"""
        SELECT a.*,b.date,b.open,b.high,b.low,b.close,b.pct_change,b.ma5,b.ma13,b.ma60,b.dif,b.dea,b.macd FROM stock_info a,stock_data b 
        where a.ts_code=b.ts_code order by a.ts_code,b.date asc
        """
        df = con.execute(query).df()
        con.close()
        select_df = pd.DataFrame()
        result=[]
        if strategy == 'dlzz':   #登陆作战
            select_df = stockUtils.get_dlzz(df)
        elif strategy == 'jtcl':   #卷土重来
            select_df = stockUtils.get_jtcl(df)
        elif strategy == 'kzjy':   #空中加油
            select_df = stockUtils.get_kzjy(df)
        elif strategy == 'yby':   #阳包阴
            select_df = stockUtils.get_bullish_engulfing_pattern(df)
        elif strategy == 'jjz':   #将军柱
            select_df = stockUtils.get_identify_general_rising_pillar_pattern(df)
        
        if not select_df.empty:
            for _, row in select_df.iterrows():
                item = {
                    'stock_info': {
                        'ts_code': str(row['ts_code']) if pd.notna(row['ts_code']) else None,
                        'stock_name': str(row['stock_name']) if pd.notna(row['stock_name']) else None,
                        'industry': str(row['industry']) if pd.notna(row['industry']) else None,
                        'gap': str(row['gap']) if pd.notna(row['gap']) else None,
                        'out_funds': float(round(row['out_funds'],2)) if pd.notna(row['out_funds']) and row['out_funds'] is not None else None,
                        'domestic': float(round(row['domestic'],2)) if pd.notna(row['domestic']) and row['domestic'] is not None else None,
                        'undervalued': float(round(row['undervalued'],2)) if pd.notna(row['undervalued']) and row['undervalued'] is not None else None,
                        'reasonable': float(round(row['reasonable'],2)) if pd.notna(row['reasonable']) and row['reasonable'] is not None else None,
                        'overvalued': float(round(row['overvalued'],2)) if pd.notna(row['overvalued']) and row['overvalued'] is not None else None,
                        'crazy': float(round(row['crazy'],2)) if pd.notna(row['crazy']) and row['crazy'] is not None else None,
                        'bg_rate': str(row['bg_rate']) if pd.notna(row['bg_rate']) else None,
                        'close': float(round(row['close'],2)) if pd.notna(row['close']) and row['close'] is not None else None,
                        'pct_change': str(row['pct_change']) if pd.notna(row['pct_change']) else None
                    }
                }
                result.append(item)

        return result
    
    async def get_gap_stocks(self):
        con = duckdb.connect(database='db/stock_prices.duckdb', read_only=False)
        # 获取今天和昨天的日期
        today = datetime.now().strftime('%Y%m%d')
        yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y%m%d')
        #上个季度
        quarter_date = stockUtils.get_latest_quarter_dates()
        # 遍历昨天和今天两天的数据
        for date in [yesterday, today]:
            df = pro.forecast(ann_date=date, type='预增', fields='ts_code,ann_date,p_change_min,p_change_max,net_profit_min,net_profit_max,last_parent_net,summary,change_reason')
            #将df中ts_code列的股票代码从'600519.SH'格式转换为'600519'格式
            df['ts_code'] = df['ts_code'].str.split('.').str[0]
            
            if not df.empty:
                #遍历df中的ts_code，用stock_utils.get_stock_name()获取股票名称
                for ts_code in df['ts_code']:
                    stock_name = stockUtils.get_stock_name(ts_code)
                    df.loc[df['ts_code'] == ts_code, 'stock_name'] = stock_name
                    #将p_change_min、p_change_max两列整合成一列p_change
                    df.loc[df['ts_code'] == ts_code, 'p_change'] = df.loc[df['ts_code'] == ts_code, 'p_change_min'].astype(str) + '-' + df.loc[df['ts_code'] == ts_code, 'p_change_max'].astype(str)
                    #将net_profit_min、net_profit_max两列整合成一列net_profit
                    df.loc[df['ts_code'] == ts_code, 'net_profit'] = df.loc[df['ts_code'] == ts_code, 'net_profit_min'].astype(str) + '-' + df.loc[df['ts_code'] == ts_code, 'net_profit_max'].astype(str)
                    #将stock_name列放到ts_code后面,删除p_change_min、p_change_max、net_profit_min、net_profit_max列
                df = df[['ts_code', 'stock_name', 'ann_date', 'p_change', 'net_profit', 'last_parent_net', 'summary', 'change_reason']]
                #gap_stock(ts_code VARCHAR, stock_name VARCHAR, ann_date VARCHAR, gap_date VARCHAR, gap_change VARCHAR, p_change VARCHAR, net_profit VARCHAR, last_parent_net DECIMAL(10, 2), summary VARCHAR, change_reason VARCHAR, "type" VARCHAR, status INTEGER DEFAULT(0));
                #将df中的数据插入到gap_stock表中
                con.register('temp_df', df)
                con.execute(f"INSERT INTO gap_stock SELECT ts_code, stock_name, ann_date, '', '', p_change, net_profit, last_parent_net, summary, change_reason, '预告',0,'{quarter_date}',0 FROM temp_df where ts_code not in (select ts_code from gap_stock where quarter_date='{quarter_date}') ")
                con.unregister('temp_df')
        
        #从gap_stock中提取check_num!=2并且status=0的记录
        query = f"""
        SELECT * FROM gap_stock WHERE check_num!=2 AND status=0 and quarter_date='{quarter_date}'
        """
        forecast_df = con.execute(query).df()
        for _,row in forecast_df.iterrows():
            check_num = row['check_num']
            ts_code = row['ts_code']
            ann_date = row['ann_date']  #发布公告日期
            if check_num == 0:
                #还没进行断层判断，根据该股票的ann_date的收盘价判断是否出现断层
                check_gap = stockUtils.get_anndate_previous_last_data(ts_code, ann_date,'1')

                if check_gap:
                    #出现断层，更新gap_stock表中check_num为1
                    con.execute(f"UPDATE gap_stock SET check_num=1,status=1 WHERE ts_code='{ts_code}' AND ann_date='{ann_date}' and quarter_date='{quarter_date}'")
                else:
                    #未出现断层，更新gap_stock表中check_num为2
                    con.execute(f"UPDATE gap_stock SET check_num=1 WHERE ts_code='{ts_code}' AND ann_date='{ann_date}' and quarter_date='{quarter_date}'")
            elif check_num == 1:
                #已判断一次，进行第二次判断，ann_date的第二天是否出现断层
                check_gap = stockUtils.get_anndate_previous_last_data(ts_code, ann_date,'2')

                if check_gap:
                    #出现跳空低开，更新gap_stock表中check_num为2
                    con.execute(f"UPDATE gap_stock SET check_num=2,status=1 WHERE ts_code='{ts_code}' AND ann_date='{ann_date}' and quarter_date='{quarter_date}'")
                else:
                    #未出现跳空低开，更新gap_stock表中check_num为1
                    con.execute(f"UPDATE gap_stock SET check_num=2 WHERE ts_code='{ts_code}' AND ann_date='{ann_date}' and quarter_date='{quarter_date}'")
                
        con.close()
        return {"status": "success"}

stockMonitorService:StockMonitorService = StockMonitorService()